Amazon S3からSnowflakeへのバルクロードを設定からやってみた
こんにちは！DA(データアナリティクス)事業本部 インテグレーション部の大高です。
SnowflakeにはAWSのS3からデータをバルクロードする仕組みがあります。以前に、事前に準備された環境でのロードはやったことがあったのですが、設定を含めてやったことがなかったので、実際にS3バケットを作成・設定するところから、SnowflakeのDBへロードするまでを試してみました。
はじめに
今回は以下のマニュアルに沿って、S3バケットの設定からデータのコピーまでの一連の流れを実際に試してみたいと思います。
Bulk Loading from Amazon S3 — Snowflake Documentation
設定
まずはS3側の設定からです。
S3へのセキュアなアクセス設定
SnowflakeからS3バケットへのアクセス設定を行う方法としては、以下のドキュメントに記載の通り3つあります。
Configuring Secure Access to Amazon S3 — Snowflake Documentation
- Snowflake ストレージ統合の設定
- IAM ロールの設定
- IAM ユーザクレデンシャルの設定
このうち、Stage の作成時やデータのロード時にIAM認証情報の提供が不要となる1番目の方法を強く推奨するとのことなので、1番目の方法で設定を行いたいと思います。
前提条件
AWS側の設定には、作業を行うAWSのIAMユーザに「IAMポリシー」と「IAMロール」の作成・管理権限が必要となるので、事前に権限を与えておきます。
Step 1: S3バケットへのアクセス権限の設定
AWSアクセスコントロール要件
S3バケットからのデータロードにおいては、下記の権限が権限が要求されます。
s3:GetObject s3:GetObjectVersion s3:ListBucket
また、もしデータをUnloadする場合には上記に加えて、下記の権限も必要となります。
s3:PutObject s3:DeleteObject
IAMポリシーの作成
上記を踏まえ、IAMポリシーを作成します。
まずはIAMの管理画面を開き、「アクセス管理 > ポリシー」から「ポリシーの作成」をクリックして、ポリシーの作成を行います。
ポリシーは「JSON」タブを開き、下記のポリシーを設定します。＜bucket＞と＜prefix＞には、実際に利用するバケット名とプレフィックス名を指定してください。指定したら「ポリシーの確認」をクリックします。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:GetObject",
"s3:GetObjectVersion",
"s3:DeleteObject",
"s3:DeleteObjectVersion"
],
"Resource": "arn:aws:s3:::<bucket>/<prefix>/*"
},
{
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": "arn:aws:s3:::<bucket>",
"Condition": {
"StringLike": {
"s3:prefix": [
"<prefix>/*"
]
}
}
}
]
}
適宜「名前」を指定して、「ポリシーの作成」をクリックします。
Step 2: IAMロールの作成
次にIAMロールを作成します。
先程と同様にIAMの管理画面を開き、「アクセス管理 > ロール」から「ロールの作成」をクリックして、ロールの作成を行います。
エンティティの種類として「別の AWS アカウント」を選択します。「アカウントID」には、一時的に、いま作業を行っているAWSアカウントのアカウントIDを指定しておきます。これは後で修正します。
アカウントIDを調べる方法は色々あると思いますが、手っ取り早いのはメニュー左下に記載されている「AWS アカウント ID」を見るのが早いかと思います。
オプションは「外部IDが必要」にチェックを入れて、「外部 ID」に0000を指定します。これも仮のIDで後で修正します。「MFAが必要」にはチェック不要です。
設定したら「次のステップ: アクセス権限」をクリックします。
ポリシーには、先程作成したポリシーを指定して、次へ進みます。
「タグ」は適宜指定して、次へ進みます。
最後に、「ロール名」に適宜名前を指定して「ロールの作成」で作成します。
Step 3: Snowflakeでのクラウドストレージ統合の作成
次に、Snowflake側でCloud Storage Integration(クラウドストレージ統合)を作成します。
なお、作成するための権限として、アカウント管理=ACCOUNTADMIN ロールまたはグローバル CREATE INTEGRATION 権限を持ったロールが必要となります。
Snowflakeにログインし、ACCOUNTADMINロールで下記クエリを実行します。＜integration_name＞には任意の名前を、＜iam_role＞には先程作成したIAMロールのARN、＜bucket＞ と ＜path＞ にはS3に作成したバケット名パスを指定します。STORAGE_ALLOWED_LOCATIONSとSTORAGE_BLOCKED_LOCATIONSは、それぞれSTORAGE INTEGRATIONから許可・拒否するパスを複数指定することができます。（STORAGE_BLOCKED_LOCATIONSはオプションです）
CREATE STORAGE INTEGRATION <integration_name>
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = S3
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '<iam_role>'
STORAGE_ALLOWED_LOCATIONS = ('s3://<bucket>/<path>/', 's3://<bucket>/<path>/')
[ STORAGE_BLOCKED_LOCATIONS = ('s3://<bucket>/<path>/', 's3://<bucket>/<path>/') ]
具体的には以下のようなクエリになります。
CREATE STORAGE INTEGRATION cm_ootaka_ext_stage_s3
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = S3
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789000:role/cm-ootaka_SnowflakeRole'
STORAGE_ALLOWED_LOCATIONS = ('s3://foobar/snowflake/load/')
これでクラウドストレージ統合が作成できました。
Step 4: SnowflakeアカウントのIAMユーザの取得
作成ができたら、以下のクエリを実行して設定値を確認します。＜integration_name＞には、先程作成した際の名前を指定します。
DESC INTEGRATION <integration_name>;
すると、下記のように出力されます。この中のSTORAGE_AWS_IAM_USER_ARNとSTORAGE_AWS_EXTERNAL_IDの値を利用して、AWS側で作成したIAMロールを修正します。
property property_type property_value property_default ENABLED Boolean true false STORAGE_ALLOWED_LOCATIONS List s3://foobar/snowflake/load/ [] STORAGE_BLOCKED_LOCATIONS List [] STORAGE_AWS_IAM_USER_ARN String arn:aws:iam::123456789000:user/abc0-d-efgh1234 STORAGE_AWS_ROLE_ARN String arn:aws:iam::123456789000:role/cm-ootaka_SnowflakeRole STORAGE_AWS_EXTERNAL_ID String FOOBAR_SFCRole=1_ABcD2eFGHIJklMNoPq3rSTUvwX4=
Step 5: IAMユーザへのバケットオブジェクトアクセス権限の付与
AWSの管理コンソールに戻り、先程作成したIAMロールを開きます。「信頼関係」タブの「信頼関係の編集」をクリックしましょう。
JSON形式でポリシードキュメントが表示されるので、Snowflake側で確認した値を設定します。＜snowflake_user_arn＞にはSTORAGE_AWS_IAM_USER_ARNの値を、＜snowflake_external_id＞にはSTORAGE_AWS_EXTERNAL_IDの値を設定します。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "",
"Effect": "Allow",
"Principal": {
"AWS": "<snowflake_user_arn>"
},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {
"sts:ExternalId": "<snowflake_external_id>"
}
}
}
]
}
設定したら「信頼ポリシーの更新」をクリックして、AWS側の設定は完了です。
Step 6: External Stageの作成
いよいよExternal Stageの作成です。Storage Integration(ストレージ統合)を参照するExternal Stageを作成します。
Snowflake側に戻り、下記のクエリを実行します。なお、事前にデータベースootaka_sandboxを作成済みです。
use schema ootaka_sandbox.public; create or replace file format cm_ootaka_csv_format type = 'CSV' field_delimiter = ',' skip_header = 1; create stage cm_ootaka_ext_stage_s3 storage_integration = cm_ootaka_ext_stage_s3 url = 's3://foobar/snowflake/load/' file_format = cm_ootaka_csv_format;
なお、今回はそのままACCOUNTADMINロールで作成していますが、通常は下記のように個別のロールに権限を与えたほうがよいです。
grant create stage on schema public to role myrole; grant usage on integration s3_int to role myrole;
これで、事前の設定準備はほぼ完了しました！
AWSにおけるデータファイルの暗号化
暗号化としては、クライアントサイド暗号化とサーバサイド暗号化があり、AWS_SSE_S3では特に設定は不要なようです。今回はAWS_SSE_S3を利用として、そのままにします。
- クライアントサイド暗号化：
- AWS_CSE: MASTER_KEY値が必要
- サーバサイド暗号化
- AWS_SSE_S3: 追加の暗号化設定は不要
- AWS_SSE_KMS: KMS_KEY_ID値の受け入れが必要
S3ステージの作成
S3ステージの作成については、前述のExternal Stageの作成を行った時点で作成済みなので、特に作業はありません！「データロード」に進みます。
データロード
S3ステージからのデータコピー
AWS側のS3の設定とSnowflake側のStageの設定は出来ているので、あとはデータとテーブルを用意しておきます。今回はシンプルに以下のようなデータとテーブルを用意します。
まずは、S3のロード用のパス（s3://foobar/snowflake/load/）に下記のファイルを配置します。
id,name,age 1,Aruto Hiden,22 2,Izu, 3,Isamu Fuwa,27 4,Yua Yaiba,24 5,Gai Amatsu,45
※完全な余談なのですが、イズ(Izu)の年齢が空欄なのは意図的です。
次に、Snowflakeのデータベースootaka_sandboxに下記のテーブルを作成します。
USE OOTAKA_SANDBOX; CREATE TABLE public.users( id INTEGER, name STRING, age INTEGER );
これで、データとテーブルの準備ができたので、下記のコマンドで実際にロードを行います。（今回は前述の通りACCOUNTADMINロールにのみ権限があるので、ACCOUNTADMINロールでCOPYコマンドを実行しています）
COPY INTO public.users FROM @cm_ootaka_ext_stage_s3;
確認してみます。
SELECT * FROM public.users;
ID NAME AGE 1 Aruto Hiden 22 2 Izu 3 Isamu Fuwa 27 4 Yua Yaiba 24 5 Gai Amatsu 45
想定どおりロードできました！
まとめ
以上、「Amazon S3からSnowflakeへバルクロードをしてみた」でした。
実際に設定からやってみると、色々と必要な作業があることが分かりました。ステップ毎に必要なセキュリティ設定をしっかり行って安全にロードできるようにしたいですね。
どなたかのお役に立てば幸いです。それでは！
